package com.bkhech.netty.rabbitmq;

import cn.hutool.setting.dialect.Props;
import com.bkhech.netty.utils.PropertiesUtil;
import com.bkhech.netty.websocket.UserChannelSession;
import com.bkhech.pojo.netty.DataContent;
import com.bkhech.utils.JsonUtils;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;


@Slf4j
public class RabbitMQConnectUtil {

    private final List<Connection> connections = new ArrayList<>();
    private final int maxConnection = 20;

    private static final Props properties;

    static {
        properties = PropertiesUtil.getInstance();
    }

    public ConnectionFactory factory;

    public ConnectionFactory getRabbitMqConnection() {
        return getFactory();
    }

    public ConnectionFactory getFactory() {
        initFactory();
        return factory;
    }

    private void initFactory() {
        try {
            if (factory == null) {
                factory = new ConnectionFactory();
                factory.setHost(properties.getProperty("rabbitmq.host"));
                factory.setPort(properties.getInt("rabbitmq.port"));
                factory.setUsername(properties.getProperty("rabbitmq.username"));
                factory.setPassword(properties.getProperty("rabbitmq.password"));
                factory.setVirtualHost(properties.getProperty("rabbitmq.virtualHost"));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void sendMsg(String message, String queue) throws Exception {
        Connection connection = getConnection();
        Channel channel = connection.createChannel();
        channel.basicPublish("",
                queue,
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                message.getBytes(StandardCharsets.UTF_8));
        channel.close();
        setConnection(connection);
    }

    public void sendMsg(String message, String exchange, String routingKey) throws Exception {
        Connection connection = getConnection();
        Channel channel = connection.createChannel();
        channel.basicPublish(exchange,
                routingKey,
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                message.getBytes(StandardCharsets.UTF_8));
        channel.close();
        setConnection(connection);
    }

    public GetResponse basicGet(String queue, boolean autoAck) throws Exception {
        GetResponse getResponse = null;
        Connection connection = getConnection();
        Channel channel = connection.createChannel();
        getResponse = channel.basicGet(queue, autoAck);
        channel.close();
        setConnection(connection);
        return getResponse;
    }

    public Connection getConnection() throws Exception {
        return getAndSetConnection(true, null);
    }

    public void setConnection(Connection connection) throws Exception {
        getAndSetConnection(false, connection);
    }

    /**
     * 监听队列
     * @param exchangeName 交换机名
     * @param queueName 队列名
     * @throws Exception
     */
    public void listen(String exchangeName, String queueName) throws Exception {
        Connection connection = getConnection();
        Channel channel = connection.createChannel();

        // 定义交换机，使用 FANOUT 发布订阅模式(广播模式)
        channel.exchangeDeclare(exchangeName,
                BuiltinExchangeType.FANOUT,
                true, false, false, null);
        // 定义队列
        channel.queueDeclare(queueName, true, false, false, null);
        // 声明绑定关系，将队列和交换机绑定，并且指定路由键
        channel.queueBind(queueName, exchangeName, "");
        // 创建消费者
        final RabbitMQMessageConsumer consumer = new RabbitMQMessageConsumer(channel, exchangeName);
        /*
         * 开启监听消息
         * queue: 监听的队列名
         * autoAck: 是否自动确认，true：告知mq消费者已经消费的确认通知
         * callback: 回调函数，处理监听到的消息
         */
        channel.basicConsume(queueName, true, consumer);
    }

    private synchronized Connection getAndSetConnection(boolean isGet, Connection connection) throws Exception {
        getRabbitMqConnection();

        if (isGet) {
            if (connections.isEmpty()) {
                return factory.newConnection();
            }
            Connection newConnection = connections.remove(0);
            if (newConnection.isOpen()) {
                return newConnection;
            } else {
                return factory.newConnection();
            }
        } else {
            if (connections.size() < maxConnection) {
                connections.add(connection);
            }
            return null;
        }
    }

}
